package zdz.stormtopo;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class SplitBolt extends BaseRichBolt {

    private OutputCollector collector;

    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector=outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {
        String sentence = tuple.getStringByField("value");
        System.out.println("------------------------------");
        System.out.println("sentence==="+sentence);
        System.out.println("offset==="+tuple.getLongByField("offset"));
        System.out.println("partition==="+tuple.getIntegerByField("partition"));
        String[] wordList = sentence.split(",");

//        for (String word : wordList){
//            System.out.println("word emit="+word);
//            collector.emit(new Values(word));
//        }
        System.out.println("word emit=" + wordList[6]);
        collector.emit(new Values(wordList[6]));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word"));
    }
}
